package test.top2.com.atguigu.app.fun;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import test.top2.com.atguigu.utils.DimUtil;
import test.top2.com.atguigu.utils.DruidDSUtil;
import test.top2.com.atguigu.utils.ThreadPoolUtil;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * ClassName: DimAsyncFunction
 * Package: test.top2.com.atguigu.app.fun
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/11 16:14
 * @Version 1.0
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements  DimJoinFunction<T>{

    private DruidDataSource druidDataSource;
    private ThreadPoolExecutor threadPoolExecutor;

    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource=DruidDSUtil.getDruidDataSource();
        threadPoolExecutor= ThreadPoolUtil.getThreadPoolExecutor();

    }

    @Override
    public void close() throws Exception {
        druidDataSource.close();
    }

    public abstract String getKey(T input);

    public  abstract void join(T input, JSONObject dimInfo);

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {

        threadPoolExecutor.execute(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {

                DruidPooledConnection connection = druidDataSource.getConnection();
                //读取维度信息
                JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, getKey(input));

                connection.close();

                //补充字段
                if (dimInfo != null){
                    join(input,dimInfo);
                }


                //将填充的字段再次写入流中
                resultFuture.complete(Collections.singleton(input));
            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("timeout:"+input);
    }
}
